package com.tdy.cdc.app.cdr;

import com.alibaba.fastjson.JSONObject;
import com.tdy.cdc.app.BaseApp;
import com.tdy.cdc.bean.ConfigBean;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.JDBCUtil;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: NanHuang
 * @Date: 2023/04/04/15:20
 * @Description: CDR层：根据配置表信息（有单表分流），异步查询MySQL相关信息，并设置状态
 */
public class CdrAppCommon extends BaseApp {
    public static void main(String[] args) {
        new CdrAppCommon().init(
                6001,
                "CdrAppCommon",
                2
        );
    }
    @Override
    public void invoke(StreamExecutionEnvironment env) {
        // 1 读取配置表信息
        HashMap<String, ConfigBean> configInfos = readConfigInfo();
        // 2 获取所有Topic
        ArrayList<String> topics = getAllTopics(configInfos);
        // 3 消费数据
        DataStreamSource<String> ds = super.readKafka(env, topics);
        // 4 处理字段映射
        SingleOutputStreamOperator<JSONObject> resultDs = processFields(ds, configInfos);
        // 5 将结果写入Kafka
        resultDs.sinkTo(FlinkSinkUtil.getKafkaSink(false,"cdr_common_apps"));
    }

    private static void processThenStr(JSONObject obj, JSONObject dwdResult, String targetColumn, String thenStr) {
        if (thenStr.contains("=")) {
            // 表达式
            if (thenStr.contains("!")) {
                String[] thenWords = thenStr.split("!=");
                String flag = thenWords[1].replaceAll("\'", "").equals("false") ? "0" : "1";
                if (!flag.equals(obj.getString(thenWords[0]))) {
                    dwdResult.put(targetColumn,"1");
                }else {
                    dwdResult.put(targetColumn,"0");
                }
            } else {
                String[] thenWords = thenStr.split("=");
                String flag = thenWords[1].replaceAll("\'", "").equals("false") ? "0" : "1";
                if (flag.equals(obj.getString(thenWords[0]))) {
                    dwdResult.put(targetColumn,"1");
                }else {
                    dwdResult.put(targetColumn,"0");
                }
            }
        } else {
            // 字段 or 具体值
            if (thenStr.contains("\'")){
                dwdResult.put(targetColumn,thenStr.replaceAll("\'",""));
            } else {
                dwdResult.put(targetColumn,obj.get(thenStr));
            }
        }
    }

    private SingleOutputStreamOperator<JSONObject> processFields(DataStreamSource<String> ds, HashMap<String, ConfigBean> configInfos) {
        /*return AsyncDataStream.unorderedWait(
                ds,
                new MyAsyncFunction(configInfos),
                120,
                TimeUnit.MINUTES
        );*/
        return ds.map(JSONObject::parseObject)
                .process(new ProcessFunction<JSONObject, JSONObject>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = JDBCUtil.getMySQLConnection();
                    }

                    @Override
                    public void processElement(JSONObject in, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject result = new JSONObject();
                        // 1 字段映射
                        String tableName = in.getString("table_name");
                        ConfigBean config = configInfos.get(tableName);
                        if (config == null) {
                            config = configInfos.get(tableName + ":" + in.getString("RECIPE_FLAG"));
                        }
                        String targetTable = config.getTargetTable();
                        result.put("table_name",targetTable);
                        result.put("op",in.getString("op"));
                        String[] sourceColumns = config.getSourceColumns().split(",");
                        String[] targetColumns = config.getTargetColumns().replaceAll(" ","").toLowerCase().split(",");
                        for (int i = 0; i < sourceColumns.length; i++) {
                            String targetColumn = targetColumns[i];
                            String sourceColumn = sourceColumns[i];
                            // 处理case-when语句
                            if (sourceColumn.contains("case") && sourceColumn.contains("when") && sourceColumn.contains("then") && sourceColumn.contains("end") ){
                                // when-then 字段
                                String regex = "when\\s+(.*?)\\s+then\\s+(.*?)\\s+(?=when|else|$)";
                                Pattern pattern = Pattern.compile(regex, Pattern.MULTILINE);
                                Matcher matcher = pattern.matcher(sourceColumn);
                                        /*
                                            本项目中case-when的格式
                                                case
                                                when 表达式1 then 表达式1/字段1/具体值1
                                                when 表达式2 then 表达式2/字段2/具体值2
                                                else 表达式/字段/具体值
                                                end

                                            注意
                                                1.表达式中出现的判断字符目前只有 "=" 或 "!="
                                                2.then语句后的表达式，一般是比较字符串值，满足则标记为1，不满足标记为0
                                        */
                                while (matcher.find()) {
                                    String whenStr = matcher.group(1).replaceAll(" ", "");
                                    String thenStr = matcher.group(2).replaceAll(" ", "");
                                    if (whenStr.contains("!")) {
                                        // 处理 "!=" 判断语句
                                        String[] words = whenStr.split("!=");
                                        if (!words[1].replaceAll("\'","").equals(in.getString(words[0]))) {
                                            // 处理then后的值
                                            processThenStr(in, result, targetColumn, thenStr);
                                        }
                                    } else {
                                        // 处理 "=" 判断语句
                                        String[] words = whenStr.split("=");
                                        if (words[1].replaceAll("\'","").equals(in.getString(words[0]))) {
                                            // 处理then后的值
                                            processThenStr(in, result, targetColumn, thenStr);
                                        }
                                    }
                                }
                                if (result.get(targetColumn) != null) {
                                    continue;
                                }
                                // else 字段
                                String elseRegex = "else\\s+(.*)end";
                                Pattern elsePattern = Pattern.compile(elseRegex, Pattern.MULTILINE);
                                Matcher elseMatcher = elsePattern.matcher(sourceColumn);
                                if (elseMatcher.find()) {
                                    String elseStr = elseMatcher.group(1);
                                    // 处理then后的值
                                    processThenStr(in, result, targetColumn, elseStr);
                                }
                            } else if (sourceColumn.contains("\'")){
                                result.put(targetColumn,sourceColumn.replaceAll("\'","").replaceAll(" ",""));
                            } else {
                                result.put(targetColumn, in.getString(sourceColumn.replaceAll(" ","").toUpperCase()));
                            }
                        }

                        if (targetTable.startsWith("cdr_hdsd")){

                            // 2 设置status
                            boolean flag1 = false;
                            boolean flag2 = false;

                            // 查询patient_id
                            String sql1 = "select GLOBAL_ID from patient where PATIENT_ID = ?";
                            PreparedStatement ps1 = conn.prepareStatement(sql1);
                            ps1.setString(1, in.getString("PATIENT_ID"));
                            ResultSet resultSet1 = ps1.executeQuery();
                            if (resultSet1.next()){
                                // patient：GLOBAL_ID -> empi
                                result.put("empi",resultSet1.getString(1));
                                flag1 = true;
                                ps1.close();
                            }

                            // 查询visit_id
                            String visitId;
                            for (String visitIdColumn : Arrays.asList("VISIT_ID","INTPATIENT_VISIT_FLOW_ID","OUTPATIENT_VISIT_FLOW_ID")) {
                                visitId = in.getString(visitIdColumn);
                                if (visitId != null && !"".equals(visitId) && !"-".equals(visitId)) {
                                    String sql2 = "select VISIT_DOMAIN_ID,VISIT_ID,PATIENT_CATEGORY from visit where VISIT_ID = ?";
                                    PreparedStatement ps2 = conn.prepareStatement(sql2);
                                    ps2.setString(1, visitId);
                                    ResultSet resultSet2 = ps2.executeQuery();
                                    if (resultSet2.next()){
                                        // visit：VisitDomainId+ "_" +VisitId + "_" + PatientCategoryCode -> encounter_id
                                        result.put("encounter_id",resultSet2.getString(1) + "_" + resultSet2.getString(2) + "_" + resultSet2.getString(3));
                                        flag2 = true;
                                        ps2.close();
                                        break;
                                    }
                                }
                            }

                            if (flag1 && flag2){
                                result.put("status","AA");
                            } else {
                                result.put("status","AE");
                            }

                        }

                        out.collect(result);
                    }

                    @Override
                    public void close() throws Exception {
                        conn.close();
                    }
                });
    }

    private ArrayList<String> getAllTopics(HashMap<String, ConfigBean> configInfos) {
        List<String> topics = new ArrayList<>(configInfos.keySet());
        // 1 去掉不存在的Topic
        Iterator<String> iterator = topics.iterator();
        while (iterator.hasNext()){
            String t = iterator.next();
            if ("dwd_mhs_source_schedule".equals(t) || "dwd_mhs_opt_appointstatus".equals(t) || "dwd_mhs_exam_appointstatus".equals(t) || "dwd_mhs_anesthesia_postoperative_record".equals(t) ||
            "dwd_mhs_surgical_nursing_record".equals(t))
                iterator.remove();
        }
        // 2 处理"表名:RECIPE_FLAG"
        return topics.stream().map(s -> {
            String[] words = s.split(":");
            return words[0];
        }).distinct().collect(Collectors.toCollection(ArrayList::new));
    }

    private HashMap<String, ConfigBean> readConfigInfo() {
        try {
            Connection conn = JDBCUtil.getConfigConnection();
            String sql = "select * from cdr_config_copy1";
            PreparedStatement ps = conn.prepareStatement(sql);
            ResultSet resultSet = ps.executeQuery();
            HashMap<String, ConfigBean> result = new HashMap<>();
            while (resultSet.next()){
                ConfigBean config = new ConfigBean(
                        resultSet.getLong(1),
                        resultSet.getString(2),
                        resultSet.getString(3),
                        resultSet.getString(4),
                        resultSet.getString(5),
                        resultSet.getString(6)
                );
                String sourceTable = config.getSourceTable();
                String extend = config.getExtend();
                if (config.getTargetTable().startsWith("cdr_hdsd00_18")){
                    continue;
                }
                if (extend == null) {
                    result.put(sourceTable,config);
                }else {
                    // 有RECIPE_FLAG拓展信息的，将"表名:RECIPE_FLAG"作为key，配置信息对象作为value
                    result.put(sourceTable + ":" + JSONObject.parseObject(extend).getString("RECIPE_FLAG"),config);
                }
            }
            ps.close();
            conn.close();
            return result;
        } catch (SQLException e) {
            throw new RuntimeException("连接MySQL中CDR配置表出错！");
        }
    }
}

